package juc;

import java.util.LinkedList;
import java.util.NoSuchElementException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer;

/**
 * 生产者&消费者-多个
 * 两个condition：生产者、消费者各一个
 * 仓库中的数据按照入库顺序消费
 * @author ben
 * @date 2021-09-08 15:08:34 CST
 */
public class ConditionExample2 {

	public static Consumer<Object> cs = System.out::println;
	// 调试
	public static Consumer<Object> dbg = System.out::println;
	
	// 以下repo、lv都是共享数据
	// 仓库：长整型
	private static final LinkedList<Long> repo = new LinkedList<>();
	// 仓库：最大容量
	private static final int CAPACITY = 10;
	// 入库数据初始值
	private static long lv = 0;
	
	// 锁
//	private static final Lock lock = new ReentrantLock();
	// ReentrantLock类中有更多 Condition的方法
//	private static final ReentrantLock lock = new ReentrantLock();
	private static final ReentrantLock lock = new ReentrantLock();
	// 仓库满：生产者阻塞
	private static final Condition FULL = lock.newCondition();
	// 仓库空：消费者阻塞
	private static final Condition EMPTY = lock.newCondition();

	/**
	 * 生产
	 * @author ben
	 * @date 2021-09-08 15:15:02 CST
	 */
	public static void produce() {
		String tname = Thread.currentThread().getName();
		final String mn = "produce";
		dbg.accept(tname + "-" + mn + ": start");
		lock.lock();
		dbg.accept(tname + "-" + mn + ": 1 EMPTY: " + lock.hasWaiters(EMPTY) + ", " + lock.getWaitQueueLength(EMPTY) 
			+ ", FULL: " + lock.hasWaiters(FULL) + ", " + lock.getWaitQueueLength(FULL));
		try {
			while (repo.size() >= CAPACITY) {
				dbg.accept(tname + "-" + mn + ": before await");
				FULL.await();
				dbg.accept(tname + "-" + mn + ": after await");
			}
			
//			TimeUnit.SECONDS.sleep(2);
			
			lv++;
			repo.addLast(lv);
			cs.accept(mn + "-" + tname + ": lv=" + lv);
			
			// 通知消费者消费
			EMPTY.signalAll();
		} catch (InterruptedException e) {
			e.printStackTrace();
		} finally {
			lock.unlock();
			dbg.accept(tname + "-" + mn + ": end");
		}
	}
	
	/**
	 * 消费
	 * @author ben
	 * @date 2021-09-08 15:15:11 CST
	 */
	public static void consume() {
		String tname = Thread.currentThread().getName();
		final String mn = "consume";
		dbg.accept(tname + "-" + mn + ": start");

		lock.lock();
		dbg.accept(tname + "-" + mn + ": 1 EMPTY: " + lock.hasWaiters(EMPTY) + ", " + lock.getWaitQueueLength(EMPTY) 
			+ ", FULL: " + lock.hasWaiters(FULL) + ", " + lock.getWaitQueueLength(FULL));
		try {
			// 必须是循环，否则，removeFirst时会出现 NoSuchElementException 错误
			while (repo.isEmpty()) {
				dbg.accept(tname + "-" + mn + ": before await");
				EMPTY.await();
				dbg.accept(tname + "-" + mn + ": after await");
			}

			TimeUnit.SECONDS.sleep(2);
			
			// 错误
//			long lvConsume = repo.getFirst();
			// pollFirst 或 removeFirst
//			long lvConsume = repo.pollFirst();
			long lvConsume = repo.removeFirst();
			cs.accept(mn + "-" + tname + ": lvConsume=" + lvConsume);
			
			// 通知生产者生产
			FULL.signalAll();
		} catch (NoSuchElementException e) {
			dbg.accept(tname + "-" + mn + ": NoSuchElementException=" + e.getMessage());
		} catch (InterruptedException e) {
			e.printStackTrace();
		} finally {
			lock.unlock();
			dbg.accept(tname + "-" + mn + ": end");
		}
	}
	
	public static void main(String[] args) {
		cs.accept("主程序开始...");
		
		ExecutorService es = Executors.newFixedThreadPool(10);
		// 写数据线程
		for (int i=0; i<5; i++) {
			es.execute(()->{
				while(true) {
					produce();
				}
			});
		}

		for (int i=0; i<5; i++) {
			// 消费者线程
			es.execute(()->{
				while(true) {
					consume();
				}
			});
		}
		
		cs.accept("主程序结束...");
	}

}
